/**
 * Copyright (c) 2013-2022 Nikita Koksharov
 * <p>
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.redisson.connection;

import org.redisson.api.NodeType;
import org.redisson.client.*;
import org.redisson.cluster.ClusterSlotRange;
import org.redisson.config.BaseConfig;
import org.redisson.config.BaseMasterSlaveServersConfig;
import org.redisson.config.MasterSlaveServersConfig;
import org.redisson.config.ReadMode;
import org.redisson.misc.RedisURI;
import org.redisson.pubsub.PublishSubscribeService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;

/**
 * ConnectionManager
 * MasterSlaveConnectionManager 类实现了 ConnectionManager 接口，
 * 用于管理主从连接的资源。
 */
public class MasterSlaveConnectionManager implements ConnectionManager {

    /**
     * 定义一个常量MAX_SLOT，表示最大的插槽数量。
     * 这个常量被声明为静态final整型，其值为16384。
     * 这个常量可以在类的任何地方直接通过类名访问，而不需要实例化对象。
     */
    public static final int MAX_SLOT = 16384;

    /**
     * 定义一个名为singleSlotRange的受保护的final变量，它是ClusterSlotRange类型的实例。
     * 这个变量表示一个单个的集群插槽范围，起始插槽编号为0，结束插槽编号为MAX_SLOT - 1。
     * ClusterSlotRange是一个处理Redis Cluster插槽范围的类。
     */
    protected final ClusterSlotRange singleSlotRange = new ClusterSlotRange(0, MAX_SLOT - 1);

    private final Logger log = LoggerFactory.getLogger(getClass());

    // DNS监控器，用于监控DNS变化
    protected DNSMonitor dnsMonitor;

    // 配置项，包含主从服务器配置
    protected MasterSlaveServersConfig config;

    // 主从服务器条目，用于存储主从服务器信息
    private MasterSlaveEntry masterSlaveEntry;

    // 发布订阅服务，用于节点间的事件通知
    protected final PublishSubscribeService subscribeService;

    // 服务管理器，用于管理各种服务的生命周期
    protected final ServiceManager serviceManager;

    // 节点连接映射，ConcurrentHashMap保证线程安全，用于存储Redis节点与其连接的映射
    private final Map<RedisURI, RedisConnection> nodeConnections = new ConcurrentHashMap<>();

    // 标记是否已连接，用于控制连接状态
    protected volatile boolean isConnected;

    // 延迟连接的 latch，使用AtomicReference封装CompletableFuture，用于异步连接初始化
    protected final AtomicReference<CompletableFuture<Void>> lazyConnectLatch = new AtomicReference<>();

    // 最后一次连接尝试
    private boolean lastAttempt;


    /**
     * 构造函数用于初始化MasterSlaveConnectionManager。
     * @param cfg 配置对象，基于BaseMasterSlaveServersConfig的泛型类型，用于配置主从服务器。
     * @param serviceManager 服务管理器，用于管理各种服务，如定时任务等。
     * 此构造函数会根据传入的配置类型，初始化配置对象config和订阅服务subscribeService，
     * 同时确保配置的合理性，例如在主从读取模式下必须定义从服务器。
     */
    public MasterSlaveConnectionManager(BaseMasterSlaveServersConfig<?> cfg, ServiceManager serviceManager) {
        this.serviceManager = serviceManager;

        // 验证配置类型并进行相应的配置初始化
        if (cfg instanceof MasterSlaveServersConfig) {
            this.config = (MasterSlaveServersConfig) cfg;
            // 如果配置了主从读取模式但未定义从服务器，则抛出异常
            if (this.config.getSlaveAddresses().isEmpty()
                    && (this.config.getReadMode() == ReadMode.SLAVE || this.config.getReadMode() == ReadMode.MASTER_SLAVE)) {
                throw new IllegalArgumentException("Slaves aren't defined. readMode can't be SLAVE or MASTER_SLAVE");
            }
        } else {
            // 如果不是MasterSlaveServersConfig类型，则通过create方法进行配置初始化
            this.config = create(cfg);
        }

        // 设置配置到服务管理器并初始化定时器
        serviceManager.setConfig(this.config);
        serviceManager.initTimer();

        // 初始化订阅服务
        subscribeService = new PublishSubscribeService(this);
    }

    @Override
    public ServiceManager getServiceManager() {
        return serviceManager;
    }

    protected void closeNodeConnections() {
        nodeConnections.values().stream()
                .map(c -> c.getRedisClient().shutdownAsync())
                .forEach(f -> f.toCompletableFuture().join());
    }

    protected void closeNodeConnection(RedisConnection conn) {
        if (nodeConnections.values().removeAll(Arrays.asList(conn))) {
            conn.closeAsync();
        }
    }

    protected final void disconnectNode(RedisURI addr) {
        RedisConnection conn = nodeConnections.remove(addr);
        if (conn != null) {
            nodeConnections.values().removeAll(Arrays.asList(conn));
            conn.closeAsync();
        }
    }

    protected final CompletionStage<RedisConnection> connectToNode(BaseConfig<?> cfg, RedisURI addr, String sslHostname) {
        return connectToNode(NodeType.MASTER, cfg, addr, sslHostname);
    }

    protected final CompletionStage<RedisConnection> connectToNode(NodeType type, BaseConfig<?> cfg, RedisURI addr, String sslHostname) {
        RedisConnection conn = nodeConnections.get(addr);
        if (conn != null) {
            if (!conn.isActive()) {
                closeNodeConnection(conn);
            } else {
                return CompletableFuture.completedFuture(conn);
            }
        }

        RedisClient client = createClient(type, addr, cfg.getConnectTimeout(), cfg.getTimeout(), sslHostname);
        CompletionStage<RedisConnection> future = client.connectAsync();
        return future.thenCompose(connection -> {
            if (connection.isActive()) {
                if (!addr.isIP()) {
                    RedisURI address = new RedisURI(addr.getScheme()
                            + "://" + connection.getRedisClient().getAddr().getAddress().getHostAddress()
                            + ":" + connection.getRedisClient().getAddr().getPort());
                    nodeConnections.put(address, connection);
                }
                nodeConnections.put(addr, connection);
                return CompletableFuture.completedFuture(connection);
            } else {
                connection.closeAsync();
                CompletableFuture<RedisConnection> f = new CompletableFuture<>();
                f.completeExceptionally(new RedisException("Connection to " + connection.getRedisClient().getAddr() + " is not active!"));
                return f;
            }
        });
    }

    @Override
    public boolean isClusterMode() {
        return false;
    }

    @Override
    public Collection<MasterSlaveEntry> getEntrySet() {
        lazyConnect();

        if (masterSlaveEntry != null) {
            return Collections.singletonList(masterSlaveEntry);
        }
        return Collections.emptyList();
    }

    protected final void lazyConnect() {
        if (!serviceManager.getCfg().isLazyInitialization()
                || isConnected) {
            return;
        }

        CompletableFuture<Void> f = new CompletableFuture<>();
        if (!lazyConnectLatch.compareAndSet(null, f)) {
            lazyConnectLatch.get().join();
            return;
        }

        try {
            connect();
            f.complete(null);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        } catch (Exception e) {
            f.completeExceptionally(e);
            lazyConnectLatch.set(null);
            throw e;
        }
    }



    /**
     * 尝试建立连接，如果首次连接失败，则根据配置进行重试。
     * 最多尝试 config.getRetryAttempts() 次。如果所有尝试都失败，将抛出异常。
     * @throws InterruptedException 如果在等待重试期间线程被中断，则抛出此异常。
     */
    public final void connect() throws InterruptedException {
        int attempts = config.getRetryAttempts() + 1; // 总共尝试的次数，包括第一次尝试
        for (int i = 0; i < attempts; i++) {
            try {
                if (i == attempts - 1) {
                    lastAttempt = true; // 标记为最后一次尝试
                }
                doConnect(); // 尝试建立连接
                return; // 成功连接后，直接返回
            } catch (Exception e) {
                if (i == attempts - 1) {
                    lastAttempt = false; // 最后一次尝试失败，重置 lastAttempt 标志
                    throw e; // 抛出捕获到的异常
                }
                try {
                    Thread.sleep(config.getRetryInterval()); // 在尝试之间进行等待
                } catch (InterruptedException ex) {
                    Thread.currentThread().interrupt(); // 保持中断状态
                    return; // 中断连接过程
                }
            }
        }
    }

    /**
     * 尝试连接到Redis服务器。此方法将根据配置决定是仅连接到单一服务器还是连接到主从服务器。
     * 如果配置为使用主从服务器，它将尝试连接到主服务器，并初始化从服务器的负载均衡。
     * 此外，还会启动DNS监控以监视服务器地址的变化。
     * @throws RuntimeException 如果连接过程中遇到运行时异常。
     * @throws RedisConnectionException 如果连接过程中遇到Redis连接相关的异常。
     */
    protected void doConnect() {
        try {
            // 根据配置决定使用单服务器入口还是主从服务器入口
            if (config.isSlaveNotUsed()) {
                masterSlaveEntry = new SingleEntry(this, serviceManager.getConnectionWatcher(), config);
            } else {
                masterSlaveEntry = new MasterSlaveEntry(this, serviceManager.getConnectionWatcher(), config);
            }

            // 设置并连接到主服务器
            CompletableFuture<RedisClient> masterFuture = masterSlaveEntry.setupMasterEntry(new RedisURI(config.getMasterAddress()));
            masterFuture.join();

            // 如果配置允许使用从服务器，则初始化从服务器负载均衡
            if (!config.isSlaveNotUsed()) {
                CompletableFuture<Void> fs = masterSlaveEntry.initSlaveBalancer(getDisconnectedNodes());
                fs.join();
            }

            // 启动DNS监控
            startDNSMonitoring(masterFuture.getNow(null));
        } catch (Exception e) {
            // 在连接出错时关闭内部资源
            internalShutdown();
            if (e instanceof CompletionException) {
                if (e.getCause() instanceof RuntimeException) {
                    throw (RuntimeException) e.getCause();
                }
                throw new RedisConnectionException(e.getCause());
            }
            throw e;
        }
    }


    protected void startDNSMonitoring(RedisClient masterHost) {
        if (masterHost.getConfig().getAddress().isIP()) {
            return;
        }

        if (config.getDnsMonitoringInterval() != -1) {
            Set<RedisURI> slaveAddresses = config.getSlaveAddresses().stream().map(r -> new RedisURI(r)).collect(Collectors.toSet());
            dnsMonitor = new DNSMonitor(this, masterHost,
                    slaveAddresses, config.getDnsMonitoringInterval(), serviceManager.getResolverGroup());
            dnsMonitor.start();
        }
    }

    protected Collection<RedisURI> getDisconnectedNodes() {
        return Collections.emptySet();
    }

    protected MasterSlaveServersConfig create(BaseMasterSlaveServersConfig<?> cfg) {
        MasterSlaveServersConfig c = new MasterSlaveServersConfig();

        // 决定了客户端发送ping消息的频率。
        c.setPingConnectionInterval(cfg.getPingConnectionInterval());

        // 启用SSL终端识别
        c.setSslEnableEndpointIdentification(cfg.isSslEnableEndpointIdentification());
        c.setSslProvider(cfg.getSslProvider());
        c.setSslTruststore(cfg.getSslTruststore());
        c.setSslTruststorePassword(cfg.getSslTruststorePassword());
        c.setSslKeystore(cfg.getSslKeystore());
        c.setSslKeystorePassword(cfg.getSslKeystorePassword());
        c.setSslProtocols(cfg.getSslProtocols());
        c.setSslCiphers(cfg.getSslCiphers());
        c.setSslKeyManagerFactory(cfg.getSslKeyManagerFactory());
        c.setSslTrustManagerFactory(cfg.getSslTrustManagerFactory());

        // 每次重试连接之间的等待时间
        c.setRetryInterval(cfg.getRetryInterval());
        // 连接失败时进行重试的次数
        c.setRetryAttempts(cfg.getRetryAttempts());
        // 单个命令执行的超时时间, 关注的是单次命令交互的及时性
        c.setTimeout(cfg.getTimeout());
        // 初始化负载均衡器为轮询算法。
        c.setLoadBalancer(cfg.getLoadBalancer());
        // 密码
        c.setPassword(cfg.getPassword());
        // 用户名
        c.setUsername(cfg.getUsername());
        // 连接名
        c.setClientName(cfg.getClientName());
        // 主节点连接池大小,默认值64
        c.setMasterConnectionPoolSize(cfg.getMasterConnectionPoolSize());
        // 从节点连接池大小,默认值64
        c.setSlaveConnectionPoolSize(cfg.getSlaveConnectionPoolSize());
        // slave节点的最大订阅（pub/sub）连接池大小。
        c.setSubscriptionConnectionPoolSize(cfg.getSubscriptionConnectionPoolSize());
        // 单个连接最大订阅数量
        c.setSubscriptionsPerConnection(cfg.getSubscriptionsPerConnection());
        // 连接超时时间，即建立连接的等待时间。如果在这段时间内无法建立连接，将会抛出异常。
        c.setConnectTimeout(cfg.getConnectTimeout());
        // 连接没有任何执行命令下空闲超,关注的是连接资源的管理
        c.setIdleConnectionTimeout(cfg.getIdleConnectionTimeout());


        // 从节点失败检测时间间隔，默认值30000毫秒
        c.setFailedSlaveCheckInterval(cfg.getFailedSlaveCheckInterval());
        // 从节点失败重连时间间隔，默认值3000毫秒
        c.setFailedSlaveReconnectionInterval(cfg.getFailedSlaveReconnectionInterval());
        // 主节点连接最小空闲大小为24
        c.setMasterConnectionMinimumIdleSize(cfg.getMasterConnectionMinimumIdleSize());
        // 从节点连接最小空闲大小为24
        c.setSlaveConnectionMinimumIdleSize(cfg.getSlaveConnectionMinimumIdleSize());
        // slave节点中，每个节点的最小空闲订阅（pub/sub）连接数量。
        c.setSubscriptionConnectionMinimumIdleSize(cfg.getSubscriptionConnectionMinimumIdleSize());
        // 默认的读取模式设置为从SLAVE读取
        c.setReadMode(cfg.getReadMode());
        // 订阅模式变量，默认为MASTER模式。
        c.setSubscriptionMode(cfg.getSubscriptionMode());
        // DNS监控的间隔时间
        c.setDnsMonitoringInterval(cfg.getDnsMonitoringInterval());
        // 连接的保活时间
        c.setKeepAlive(cfg.isKeepAlive());
        // TCP_NODELAY是一个TCP/IP选项
        c.setTcpNoDelay(cfg.isTcpNoDelay());
        c.setNameMapper(cfg.getNameMapper());
        // Redis连接的用户凭证信息
        c.setCredentialsResolver(cfg.getCredentialsResolver());
        c.setCommandMapper(cfg.getCommandMapper());

        return c;
    }

    @Override
    public RedisClient createClient(NodeType type, RedisURI address, String sslHostname) {
        RedisClient client = createClient(type, address, config.getConnectTimeout(), config.getTimeout(), sslHostname);
        return client;
    }

    @Override
    public RedisClient createClient(NodeType type, InetSocketAddress address, RedisURI uri, String sslHostname) {
        RedisClient client = createClient(type, address, uri, config.getConnectTimeout(), config.getTimeout(), sslHostname);
        return client;
    }

    protected RedisClient createClient(NodeType type, RedisURI address, int timeout, int commandTimeout, String sslHostname) {
        RedisClientConfig redisConfig = createRedisConfig(type, address, timeout, commandTimeout, sslHostname);
        return RedisClient.create(redisConfig);
    }

    private RedisClient createClient(NodeType type, InetSocketAddress address, RedisURI uri, int timeout, int commandTimeout, String sslHostname) {
        RedisClientConfig redisConfig = createRedisConfig(type, null, timeout, commandTimeout, sslHostname);
        redisConfig.setAddress(address, uri);
        return RedisClient.create(redisConfig);
    }

    protected RedisClientConfig createRedisConfig(NodeType type, RedisURI address, int timeout, int commandTimeout, String sslHostname) {
        RedisClientConfig redisConfig = new RedisClientConfig();
        redisConfig.setAddress(address)
                .setTimer(serviceManager.getTimer())
                .setExecutor(serviceManager.getExecutor())
                .setResolverGroup(serviceManager.getResolverGroup())
                .setGroup(serviceManager.getGroup())
                .setSocketChannelClass(serviceManager.getSocketChannelClass())
                .setConnectTimeout(timeout)
                .setCommandTimeout(commandTimeout)
                .setSslHostname(sslHostname)
                .setSslEnableEndpointIdentification(config.isSslEnableEndpointIdentification())
                .setSslProvider(config.getSslProvider())
                .setSslTruststore(config.getSslTruststore())
                .setSslTruststorePassword(config.getSslTruststorePassword())
                .setSslKeystore(config.getSslKeystore())
                .setSslKeystorePassword(config.getSslKeystorePassword())
                .setSslProtocols(config.getSslProtocols())
                .setSslCiphers(config.getSslCiphers())
                .setSslKeyManagerFactory(config.getSslKeyManagerFactory())
                .setSslTrustManagerFactory(config.getSslTrustManagerFactory())
                .setClientName(config.getClientName())
                .setKeepPubSubOrder(serviceManager.getCfg().isKeepPubSubOrder())
                .setPingConnectionInterval(config.getPingConnectionInterval())
                .setKeepAlive(config.isKeepAlive())
                .setTcpNoDelay(config.isTcpNoDelay())
                .setUsername(config.getUsername())
                .setPassword(config.getPassword())
                .setNettyHook(serviceManager.getCfg().getNettyHook())
                .setCommandMapper(config.getCommandMapper())
                .setCredentialsResolver(config.getCredentialsResolver())
                .setConnectedListener(addr -> {
                    isConnected = true;
                    if (!serviceManager.isShuttingDown()) {
                        NodeType nt = getNodeType(type, addr);
                        serviceManager.getConnectionEventsHub().fireConnect(addr, nt);
                    }
                })
                .setDisconnectedListener(addr -> {
                    if (!serviceManager.isShuttingDown()) {
                        NodeType nt = getNodeType(type, addr);
                        serviceManager.getConnectionEventsHub().fireDisconnect(addr, nt);
                    }
                });

        if (type != NodeType.SENTINEL) {
            redisConfig.setDatabase(config.getDatabase());
        }

        return redisConfig;
    }

    private NodeType getNodeType(NodeType type, InetSocketAddress address) {
        if (type != NodeType.SENTINEL) {
            MasterSlaveEntry entry = getEntry(address);
            if (entry != null) {
                if (!entry.isInit()) {
                    return type;
                }
                InetSocketAddress addr = entry.getClient().getAddr();
                if (addr.getAddress().equals(address.getAddress())
                        && addr.getPort() == address.getPort()) {
                    return NodeType.MASTER;
                }
            }
            return NodeType.SLAVE;
        }
        return type;
    }

    @Override
    public int calcSlot(String key) {
        return singleSlotRange.getStartSlot();
    }

    @Override
    public int calcSlot(byte[] key) {
        return singleSlotRange.getStartSlot();
    }

    @Override
    public MasterSlaveEntry getEntry(InetSocketAddress address) {
        lazyConnect();

        return masterSlaveEntry;
    }

    @Override
    public MasterSlaveEntry getEntry(RedisURI addr) {
        lazyConnect();

        return masterSlaveEntry;
    }

    @Override
    public MasterSlaveEntry getEntry(RedisClient redisClient) {
        lazyConnect();

        return masterSlaveEntry;
    }

    @Override
    public MasterSlaveEntry getEntry(String name) {
        int slot = calcSlot(name);
        return getEntry(slot);
    }

    protected MasterSlaveEntry getEntry(int slot) {
        lazyConnect();

        return masterSlaveEntry;
    }

    @Override
    public MasterSlaveEntry getWriteEntry(int slot) {
        return getEntry(slot);
    }

    @Override
    public MasterSlaveEntry getReadEntry(int slot) {
        return getEntry(slot);
    }

    protected CompletableFuture<RedisClient> changeMaster(int slot, RedisURI address) {
        MasterSlaveEntry entry = getEntry(slot);
        return entry.changeMaster(address);
    }

    protected void internalShutdown() {
        if (lazyConnectLatch.get() == null && lastAttempt) {
            shutdown();
        }
    }

    @Override
    public void shutdown() {
        shutdown(0, 2, TimeUnit.SECONDS); //default netty value
    }

    @Override
    public void shutdown(long quietPeriod, long timeout, TimeUnit unit) {
        if (dnsMonitor != null) {
            dnsMonitor.stop();
        }

        serviceManager.getConnectionWatcher().stop();

        if (isConnected) {
            List<CompletableFuture<Void>> futures = new ArrayList<>();
            for (MasterSlaveEntry entry : getEntrySet()) {
                futures.add(entry.shutdownAsync());
            }
            CompletableFuture<Void> future = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));

            try {
                future.get(timeout, unit);
            } catch (Exception e) {
                // skip
            }
        }

        serviceManager.getResolverGroup().close();

        serviceManager.getShutdownLatch().close();
        if (serviceManager.getCfg().getExecutor() == null) {
            serviceManager.getExecutor().shutdown();
            try {
                serviceManager.getExecutor().awaitTermination(timeout, unit);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }

        serviceManager.getShutdownPromise().trySuccess(null);
        serviceManager.getShutdownLatch().awaitUninterruptibly();

        if (serviceManager.getCfg().getEventLoopGroup() == null) {
            serviceManager.getGroup().shutdownGracefully(quietPeriod, timeout, unit).syncUninterruptibly();
        }

        serviceManager.getTimer().stop();
    }

    @Override
    public PublishSubscribeService getSubscribeService() {
        return subscribeService;
    }

    @Override
    public RedisURI getLastClusterNode() {
        return null;
    }

}
